package com.wenx.jettyCamel;

import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.ReadContext;
import net.sf.json.JSONArray;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.camel.component.ActiveMQComponent;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.model.RouteDefinition;

import javax.jms.ConnectionFactory;
import javax.jms.Session;

import static org.apache.camel.builder.Builder.constant;

public class JettyToMQConsumer {
    private static String user = "admin";
    private static String password = "admin";
    //填写AMQ的IP
    private static String url = "tcp://10.10.3.186:61616";
    public static void main(String[] args) throws Exception {
        CamelContext context = new DefaultCamelContext(); // 1. 创建 CamelContext.
        //创建AMQ的连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
        ActiveMQComponent activeMQComponent = new ActiveMQComponent();
        activeMQComponent.setConnectionFactory(connectionFactory);
        //使用连接池连接
        activeMQComponent.setUsePooledConnection(true);
        //事务
        activeMQComponent.setTransacted(true);
        //客户端确认
        activeMQComponent.setAcknowledgementMode(Session.CLIENT_ACKNOWLEDGE);
        //持久化
        activeMQComponent.setDeliveryPersistent(true);
        context.addComponent("activemq", activeMQComponent);
        System.out.println(url + " " + user + " " + password);
        context.setTracing(true);
        context.start(); // 4. 启动CamelContext.
        context.addRouteDefinition(initRoute3());
        Thread.sleep(Integer.MAX_VALUE);  // 为了保持CamelContext处于工作状态，这里需要sleep主线程
        context.stop(); // 最后停止CamelContext
    }
    private static int count = 1;
    public static RouteDefinition initRoute3(){
        RouteDefinition routeDefinition = new RouteDefinition();
        routeDefinition.from("activemq:queue:Consumer.A.VirtualTopic.tgb")
                .process(new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        String body = exchange.getIn().getBody(String.class);
                        System.out.println("第"+count+"次消费的body="+body);
                        exchange.getOut().setBody(body);
                        count++;
                    }
                })
                .setHeader(Exchange.HTTP_METHOD, constant("POST"))
                .setHeader(Exchange.CONTENT_TYPE,constant("application/json"))
                .to("http://localhost:8081/demo/hello7");
        return routeDefinition;
    }
}
